package cn.ac.iscas.dao.mapper;

import cn.ac.iscas.dao.entity.FlinkJobHistoryDO;
import cn.ac.iscas.domain.job.Job;
import com.baomidou.dynamic.datasource.annotation.DS;
import org.apache.ibatis.annotations.*;
import org.apache.ibatis.type.JdbcType;

//TODO: 内部使用数据库切换
@Mapper
//使用mysql数据库
//@DS("master")
//使用达梦数据库
@DS("dameng")
public interface IFlinkJobHistoryMapper
{
    @Select({
            "SELECT a.`jobId`, a.`flinkjobId`, a.`executeTime`, a.`executeCount`, a.`status`, a.`duration` " +
                    " FROM flink_job_history AS a " +
                    " LEFT JOIN jobmetadata AS b ON a.`jobId` = b.`_id` " +
                    " WHERE a.`jobId` = #{jobId,jdbcType=CHAR} " +
                    " ORDER BY a.`executeTime` DESC " +
                    " LIMIT 1 "
    })
    @Results({
            @Result(column="jobId", property="jobId", jdbcType= JdbcType.VARCHAR, id=true),
            @Result(column="flinkjobId", property="flinkjobId", jdbcType=JdbcType.VARCHAR),
            @Result(column="executeTime", property="executeTime", jdbcType=JdbcType.DATE),
            @Result(column="status", property="status", jdbcType=JdbcType.VARCHAR),
            @Result(column="executeCount", property="executeCount", jdbcType=JdbcType.INTEGER),
            @Result(column="duration", property="duration", jdbcType=JdbcType.BIGINT)
    })
    FlinkJobHistoryDO selectLatestByJobId(String jobId);

    @Insert({
            "INSERT INTO " +
                    " flink_job_history " +
                    " (`jobId`, `flinkJobId`, `executeTime`, `executeCount`, `status`, `duration` ) ",
            " VALUES (#{jobId,jdbcType=CHAR}, #{flinkJobId,jdbcType=VARCHAR}, ",
                    " #{executeTime,jdbcType=VARCHAR}, #{executeCount,jdbcType=VARCHAR}, ",
                    " #{status,jdbcType=VARCHAR}, #{duration,jdbcType=BIGINT}) "
    })
    int insert(FlinkJobHistoryDO job);

    @Select({
            "SELECT executeCount FROM flink_job_history WHERE jobId = #{jobId,jdbcType=CHAR} ORDER BY executeCount DESC LIMIT 1;"
    })
    int selectMaxExecuteCoundByJobId(String jobId);

    @Select({
            "SELECT COUNT(jobId) FROM flink_job_history WHERE jobId = #{jobId,jdbcType=CHAR} ORDER BY executeCount DESC LIMIT 1;"
    })
    int selectJobIdCount(String jobId);

    @Update({
            "UPDATE " +
                    " flink_job_history " +
                    " SET status=#{status, jdbcType=VARCHAR} " +
                    " WHERE " +
                    " flinkJobId=#{flinkJobId, jdbcType=VARCHAR}"

    })
    int updateStatus(String flinkJobId, String status);

    @Update({
            "UPDATE " +
                    " flink_job_history " +
                    " SET status=#{status, jdbcType=VARCHAR}, " +
                    " duration=#{duration, jdbcType=BIGINT} " +
                    " WHERE " +
                    " flinkJobId=#{flinkJobId, jdbcType=VARCHAR}"

    })
    int updateStatusDuration(String flinkJobId, String status, Long duration);
}
